package org.apache.ocean;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import org.apache.lucene.search.Query;


/**
 *
 * @author jasonr
 */
public class MultiShardFacetCount {
  List<SearchIndexShard> searchIndexShards;
  ThreadPoolExecutor threadPool;
  TimeoutState timeoutState;
  
  public MultiShardFacetCount(List<SearchIndexShard> searchIndexShards, ThreadPoolExecutor threadPool, TimeoutState timeoutState) {
    this.searchIndexShards = searchIndexShards;
    this.threadPool = threadPool;
    this.timeoutState = timeoutState;
  }
  
  public CLong getFacetCount(Query a, Query b) throws Throwable {
    List<Callable<CLong>> tasks = new ArrayList<Callable<CLong>>(searchIndexShards.size());
    for (SearchIndexShard searchIndexShard : searchIndexShards) {
      tasks.add(new FacetCountTask(a, b, searchIndexShard));
    }
    List<Future<CLong>> futures = threadPool.invokeAll(tasks);
    CLong total = new CLong(0);
    for (Future<CLong> future : futures) {
      try {
        CLong value = future.get();
        total.value += value.value;
      } catch (ExecutionException executionException) {
        timeoutState.cancel(executionException.getCause());
        throw executionException.getCause();
      }
    }
    return total;
  }
  
  public class FacetCountTask implements Callable<CLong> {
    Query a;
    Query b;
    SearchIndexShard searchIndexShard;
    
    public FacetCountTask(Query a, Query b, SearchIndexShard searchIndexShard) {
      this.a = a;
      this.b = b;
      this.searchIndexShard = searchIndexShard;
    }
    
    public CLong call() throws Exception {
      CLong total = new CLong(0);
      /**
      List<IndexSearcher> indexSearchers = searchIndexShard.createIndexSearchers();
      for (IndexSearcher indexSearcher : indexSearchers) {
        IndexReader indexReader = indexSearcher.getIndexReader();
        CLong count = indexReader.getFacetCount(a, b, timeoutState);
        total.value += count.value;
      }
      **/
      return total;
    }
  }
}
